package com.joker.demo.core.netty.client;

import com.alibaba.fastjson2.JSON;
import com.joker.demo.core.config.JobExecutorProperties;
import com.joker.demo.core.domain.bo.JobRegistryBO;
import com.joker.demo.core.domain.dto.MessageDTO;
import com.joker.demo.core.domain.enums.JobRequestEnum;
import com.joker.demo.core.exception.BizException;
import com.joker.demo.core.executor.JobExecutor;
import com.joker.demo.core.netty.client.handler.ClientProtocolRouterHandler;
import com.joker.demo.core.netty.client.handler.NettyClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.json.JsonObjectDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty客户端管理器
 *
 * @author feixiang.li
 * @since 2025/7/27 16:09
 */
public class NettyClient {

    private static final Logger log = LoggerFactory.getLogger(NettyClient.class);

    private final JobExecutorProperties config;
    private final NettyClientHandler clientHandler;
    private Channel clientChannel;
    private EventLoopGroup clientGroup;
    private ScheduledExecutorService heartbeatExecutor;
    private boolean isConnected = false;
    private final JobExecutor jobExecutor;
    
    // 重试相关
    private ScheduledExecutorService retryExecutor;
    private final AtomicInteger retryCount = new AtomicInteger(0);
    private static final int MAX_RETRY_COUNT = 3;
    private static final int RETRY_INTERVAL_SECONDS = 10;
    private JobRegistryBO lastRegisteredJobRegistryBO;

    public NettyClient(JobExecutorProperties config, NettyClientHandler clientHandler, JobExecutor jobExecutor) {
        this.config = config;
        this.clientHandler = clientHandler;
        this.jobExecutor = jobExecutor;
        // 设置重连回调
        this.clientHandler.setReconnectCallback(this::handleReconnect);
    }

    /**
     * 连接到调度器
     */
    public void connect() {
        log.info("连接到调度器: {}", config.getServerUrl());
        log.info("本机启动的端口: {}", config.getPort());
        clientGroup = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(clientGroup)
                    .channel(NioSocketChannel.class)
                    // 设置本地绑定端口
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            // 消息类型检测器
                            p.addLast(new ClientProtocolRouterHandler());
                            // json 解码器-入栈
                            p.addLast(new JsonObjectDecoder());
                            // 入栈 - 指定UTF-8编码
                            p.addLast(new StringDecoder(CharsetUtil.UTF_8));
                            // 出栈 - 指定UTF-8编码
                            p.addLast(new StringEncoder(CharsetUtil.UTF_8));
                            p.addLast(clientHandler);
                            // log 是双端处理
                            p.addLast(new LoggingHandler(LogLevel.DEBUG));
                        }
                    });

            // 解析调度器地址
            String host = config.getServerUrl().replace("http://", "");
            int port = config.getServerPort();
            log.info("解析到的调度器地址: {}:{}", host, port);
            clientChannel = bootstrap.connect(host, port).sync().channel();
            isConnected = true;
            // 重置重试计数器
            retryCount.set(0);
            log.info("成功连接到调度器: {}:{}", host, port);
        } catch (InterruptedException e) {
            log.error("连接调度器失败", e);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("连接调度器失败", e);
            throw new BizException("连接调度器失败", e);
        }
    }

    /**
     * 处理重连
     */
    private void handleReconnect() {
        log.info("检测到连接断开，开始重连机制");
        
        if (retryExecutor == null) {
            retryExecutor = Executors.newSingleThreadScheduledExecutor();
        }
        
        // 停止心跳
        if (heartbeatExecutor != null) {
            heartbeatExecutor.shutdown();
            heartbeatExecutor = null;
        }
        
        // 标记连接断开
        isConnected = false;
        
        // 启动重试任务
        retryExecutor.schedule(() -> {
            try {
                retryConnect();
            } catch (Exception e) {
                log.error("重连过程中发生异常", e);
            }
        }, RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
    }
    
    /**
     * 重试连接
     */
    private void retryConnect() {
        int currentRetryCount = retryCount.incrementAndGet();
        
        if (currentRetryCount > MAX_RETRY_COUNT) {
            log.error("重连失败，已达到最大重试次数: {}", MAX_RETRY_COUNT);
            return;
        }
        
        log.info("开始第 {} 次重连尝试", currentRetryCount);
        
        try {
            // 关闭旧的连接
            if (clientChannel != null) {
                clientChannel.close();
                clientChannel = null;
            }
            
            // 重新连接
            connect();
            
            // 重新注册任务处理器
            if (lastRegisteredJobRegistryBO != null) {
                registerJobHandlers(lastRegisteredJobRegistryBO);
            }
            
            // 重新启动心跳
            startHeartbeat();
            
            log.info("重连成功！");
            
        } catch (Exception e) {
            log.error("第 {} 次重连失败", currentRetryCount, e);
            
            // 如果还有重试次数，继续重试
            if (currentRetryCount < MAX_RETRY_COUNT) {
                retryExecutor.schedule(() -> {
                    try {
                        retryConnect();
                    } catch (Exception ex) {
                        log.error("重连过程中发生异常", ex);
                    }
                }, RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
            } else {
                log.error("重连失败，已达到最大重试次数: {}", MAX_RETRY_COUNT);
            }
        }
    }

    /**
     * 注册任务处理器
     */
    public void registerJobHandlers(JobRegistryBO jobRegistryBO) {
        if (!isConnected) {
            throw new BizException("客户端未连接到调度器");
        }

        log.info("开始注册任务处理器");
        
        // 保存注册信息，用于重连时重新注册
        this.lastRegisteredJobRegistryBO = jobRegistryBO;

        // 发送注册消息
        MessageDTO registerMessageDTO = new MessageDTO(JobRequestEnum.REGISTER.getCode(), UUID.randomUUID().toString(), JSON.toJSONString(jobRegistryBO));
        clientChannel.writeAndFlush(JSON.toJSONString(registerMessageDTO));

        log.info("任务处理器注册完成，共注册 {} 个处理器", jobRegistryBO.getJobRegistryList().size());
    }

    /**
     * 启动心跳机制
     */
    public void startHeartbeat() {
        log.info("启动心跳机制");
        if (!isConnected) {
            throw new BizException("客户端未连接到调度器");
        }

        log.info("启动心跳机制");

        heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            try {
                MessageDTO heartbeatMessageDTO = new MessageDTO(JobRequestEnum.HEARTBEAT.getCode(), UUID.randomUUID().toString(),
                        JSON.toJSONString(config.getGroupName()));
                clientChannel.writeAndFlush(JSON.toJSONString(heartbeatMessageDTO));
                log.info("发送心跳消息");
            } catch (Exception e) {
                log.error("发送心跳消息失败", e);
            }
        }, 10, 10, TimeUnit.SECONDS);
    }

    /**
     * 关闭连接
     */
    public void disconnect() {
        log.info("关闭Netty客户端连接");

        if (heartbeatExecutor != null) {
            heartbeatExecutor.shutdown();
        }

        if (retryExecutor != null) {
            retryExecutor.shutdown();
        }

        if (clientChannel != null) {
            clientChannel.close();
        }

        if (clientGroup != null) {
            clientGroup.shutdownGracefully();
        }

        isConnected = false;
        log.info("Netty客户端连接已关闭");
    }

    /**
     * 检查连接状态
     */
    public boolean isConnected() {
        return isConnected && clientChannel != null && clientChannel.isActive();
    }

    /**
     * 发送消息
     */
    public void sendMessage(MessageDTO messageDTO) {
        if (!isConnected) {
            throw new BizException("客户端未连接到调度器");
        }

        clientChannel.writeAndFlush(JSON.toJSONString(messageDTO));
    }
} 